//
// TcpServerDispatcher.cpp
//
// $Id: //poco/1.4/Net/src/TcpServerDispatcher.cpp#1 $
//
// Library: Net
// Package: TcpServer
// Module:  TcpServerDispatcher
//
// Copyright (c) 2005-2007, Applied Informatics Software Engineering GmbH.
// and Contributors.
//
// SPDX-License-Identifier:	BSL-1.0
//


#include "Poco/Net/TcpServerDispatcher.h"
#include "Poco/Net/TcpServerConnectionFactory.h"
#include "Poco/Notification.h"
#include <memory>


using Poco::Notification;
using Poco::FastMutex;


namespace Poco {
namespace Net {


class TcpConnectionNotification: public Notification
{
public:
	TcpConnectionNotification(int fd):
		m_fd(fd)
	{
	}
	
	~TcpConnectionNotification()
	{
	}
	
	int fd() const
	{
		return m_fd;
	}

private:
	int m_fd;
};


TcpServerDispatcher::TcpServerDispatcher(TcpServerConnectionFactory* pFactory, Poco::ThreadPool& threadPool, TcpServerParams* pParams):
	m_rc(1),
	m_pParams(pParams),
	m_currentThreads(0),
	m_totalConnections(0),
	m_currentConnections(0),
	m_maxConcurrentConnections(0),
	m_refusedConnections(0),
	m_stopped(false),
	m_pConnectionFactory(pFactory),
	m_threadPool(threadPool)
{
	poco_check_ptr (pFactory);

	if (!m_pParams)
		m_pParams = new TcpServerParams;
	
	if (m_pParams->getMaxThreads() == 0)
		m_pParams->setMaxThreads(threadPool.capacity());
}


TcpServerDispatcher::~TcpServerDispatcher()
{
}


void TcpServerDispatcher::duplicate()
{
	m_mutex.lock();
	++m_rc;
	m_mutex.unlock();
}


void TcpServerDispatcher::release()
{
	m_mutex.lock();
	int rc = --m_rc;
	m_mutex.unlock();
	if (rc == 0) delete this;
}


void TcpServerDispatcher::run()
{
	int idleTime = (int) m_pParams->getThreadIdleTime().totalMilliseconds();

	for (;;)
	{
		Notification* pNf = m_queue.waitDequeueNotification(idleTime);
		if (pNf)
		{
			TcpConnectionNotification* pCNf = dynamic_cast<TcpConnectionNotification*>(pNf);
			if (pCNf)
			{
				std::auto_ptr<TcpServerConnection> pConnection(m_pConnectionFactory->createConnection(pCNf->fd()));
				poco_check_ptr(pConnection.get());
				beginConnection();
				pConnection->start();
				endConnection();
			}
		}
	
		FastMutex::ScopedLock lock(m_mutex);
		if (m_stopped || (m_currentThreads > 1 && m_queue.empty()))
		{
			--m_currentThreads;
			break;
		}
	}
}


namespace
{
	static const std::string threadName("TcpServerConnection");
}

	
void TcpServerDispatcher::enqueue(int fd)
{
	FastMutex::ScopedLock lock(m_mutex);

	if (m_queue.size() < m_pParams->getMaxQueued())
	{
		m_queue.enqueueNotification(new TcpConnectionNotification(fd));
		if (!m_queue.hasIdleThreads() && m_currentThreads < m_pParams->getMaxThreads())
		{
			try
			{
				m_threadPool.startWithPriority(m_pParams->getThreadPriority(), *this, threadName);
				++m_currentThreads;
			}
			catch (Poco::Exception&)
			{
				// no problem here, connection is already queued
				// and a new thread might be available later.
			}
		}
	}
	else
	{
		++m_refusedConnections;
	}
}


void TcpServerDispatcher::stop()
{
	m_stopped = true;
	m_queue.clear();
	m_queue.wakeUpAll();
}


int TcpServerDispatcher::currentThreads() const
{
	FastMutex::ScopedLock lock(m_mutex);
	
	return m_currentThreads;
}

int TcpServerDispatcher::maxThreads() const
{
	FastMutex::ScopedLock lock(m_mutex);
	
	return m_threadPool.capacity();
}


int TcpServerDispatcher::totalConnections() const
{
	FastMutex::ScopedLock lock(m_mutex);
	
	return m_totalConnections;
}


int TcpServerDispatcher::currentConnections() const
{
	FastMutex::ScopedLock lock(m_mutex);
	
	return m_currentConnections;
}


int TcpServerDispatcher::maxConcurrentConnections() const
{
	FastMutex::ScopedLock lock(m_mutex);
	
	return m_maxConcurrentConnections;
}


int TcpServerDispatcher::queuedConnections() const
{
	return m_queue.size();
}


int TcpServerDispatcher::refusedConnections() const
{
	FastMutex::ScopedLock lock(m_mutex);
	
	return m_refusedConnections;
}


void TcpServerDispatcher::beginConnection()
{
	FastMutex::ScopedLock lock(m_mutex);
	
	++m_totalConnections;
	++m_currentConnections;
	if (m_currentConnections > m_maxConcurrentConnections)
		m_maxConcurrentConnections = m_currentConnections;
}


void TcpServerDispatcher::endConnection()
{
	FastMutex::ScopedLock lock(m_mutex);

	--m_currentConnections;
}


} } // namespace Poco::Net
